package javaVersion.learn.stateProcess.keyStateDemo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 *
 *
 * 实现功能，有一组数据，
 *     (1L, 3L),
 *     (1L, 5L),
 *     (1L, 7L),
 *     (1L, 4L),
 *     (1L, 2L),
 *     (1L, 4L)
 *  对相同键的值保存状态并计算 每两条数据值（value）的平均值，并输出。
 *      3+5 = 8      8/2=4             （1，4）
 *      7+4 = 11     11/2=5             (1，5)
 *      2+4 = 6      6/2=3              (1，3)
 *      最终输出结果：  (1，4)
 *                      (1，5)
 *                      (1，3)
 * @program: myFlink
 * @description: 窗口平均计数案例
 * @author: WincoMa
 * @create: 2020-07-24 17:24
 **/
public class KeyState_ExampleCountWindowAverage {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        ArrayList<Tuple2<Long,Long>> list = new ArrayList<>();
        Tuple2<Long, Long> e1 = new Tuple2<>(1L,3L);
        Tuple2<Long, Long> e2 = new Tuple2<>(1L,5L);
        Tuple2<Long, Long> e3 = new Tuple2<>(1L,7L);
        Tuple2<Long, Long> e4 = new Tuple2<>(1L,4L);
        Tuple2<Long, Long> e5 = new Tuple2<>(1L,2L);
        Tuple2<Long, Long> e6 = new Tuple2<>(1L,4L);
        list.add(e1);
        list.add(e2);
        list.add(e3);
        list.add(e4);
        list.add(e5);
        list.add(e6);
        senv.fromCollection(list)
                .keyBy(0)
                .flatMap(new KeyState_StateManager())
                .print();
        senv.execute("ExampleManagedState");
    }
}
